package com.learn.kafka.assign;


import com.learn.kafka.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.time.Duration;
import java.util.Arrays;

/**
 * 演示分区自动分配策略的消费者（继承通用配置父类）
 * @author: learn.kafka
 */
public class ConsumerAutoAssignTest extends ConsumerProperties {
    public static void main(String[] args) {

        new ConsumerAutoAssignTest().startConsuming();
        // 结果 c1 : p0 p1 p2  c2 : p3 p4 （默认）
        // props.put("partition.assignment.strategy","org.apache.kafka.clients.consumer.RangeAssignor");

        // 结果 c1 : p0 p2 p4  c2 : p1 p3
        //props.put("partition.assignment.strategy","org.apache.kafka.clients.consumer.RoundRobinAssignor");

        // 结果 c1 : p2 p3 p4  c2 : p0 p1
        //props.put("partition.assignment.strategy","org.apache.kafka.clients.consumer.StickyAssignor");

    }

    // 构造方法：初始化分区分配相关配置
    public ConsumerAutoAssignTest() {
        super(); // 调用父类构造方法，加载通用配置
        initAssignConfig();
    }

    // 初始化分区分配策略等特殊配置
    private void initAssignConfig() {
        // 设置消费组ID（覆盖父类默认值）
        setProperty("group.id", "mate-assign-group");

        // 可选：配置分区分配策略（默认是RangeAssignor）
        // 1. 范围分配策略（默认）：c1消费p0-p2，c2消费p3-p4
        // setConfig("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RangeAssignor");

        // 2. 轮询分配策略：c1消费p0、p2、p4，c2消费p1、p3
        // setConfig("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");

        // 3. 粘性分配策略：尽量保持现有分配，仅在必要时最小化调整
        // setConfig("partition.assignment.strategy", "org.apache.kafka.clients.consumer.StickyAssignor");
    }

    // 启动两个消费者，演示分区自动分配
    private void startConsuming() {
        // 从父类获取两个消费者实例（同一消费组）
        Consumer<Integer, String> consumer1 = createConsumer();
        Consumer<Integer, String> consumer2 = createConsumer();

        // 订阅主题（两个消费者订阅同一主题）
        consumer1.subscribe(Arrays.asList("ass5part"));
        consumer2.subscribe(Arrays.asList("ass5part"));

        try {
            // 持续消费消息
            while (true) {
                // 消费者1拉取并处理消息
                ConsumerRecords<Integer, String> records1 = consumer1.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<Integer, String> record : records1) {
                    System.out.printf(
                            "----consume1----offset = %d, key = %s, value = %s, partition = %d%n",
                            record.offset(), record.key(), record.value(), record.partition()
                    );
                }

                // 消费者2拉取并处理消息
                ConsumerRecords<Integer, String> records2 = consumer2.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<Integer, String> record : records2) {
                    System.out.printf(
                            "----consume2----offset = %d, key = %s, value = %s, partition = %d%n",
                            record.offset(), record.key(), record.value(), record.partition()
                    );
                }
            }
        } finally {
            // 关闭消费者
            consumer1.close();
            consumer2.close();
        }
    }
}
